#include <iostream>
#include <thread>
#include "rxcpp/rx.hpp"

namespace rx
{
    using namespace rxcpp;
    using namespace rxcpp::sources;
    using namespace rxcpp::operators;
    using namespace rxcpp::util;
}
using namespace rx;

void fonnext(int v)
{
    std::cout << "observer next() thread Id: " << std::this_thread::get_id() << std::endl;
    std::cout << "value " << v << std::endl;
}

void foncomplete()
{
    std::cout << "observer complete() thread Id: " << std::this_thread::get_id() << std::endl;
}

void fobservable(rxcpp::subscriber<int> s)
{
    s.on_next(1);
    s.on_next(2);
    s.on_completed();
}

int main(int, char **)
{
    std::cout << "Hello, rxcpp!\n";

    auto int$ = rxcpp::observable<>::create<int>(fobservable);
    int$.subscribe(
        [](int v)
        { printf("int$ OnNext: %d\n ", v); },
        []()
        { printf("int$ OnCompleted\n"); });

    auto threads = rxcpp::observe_on_event_loop();
    auto value$ = rx::observable<>::range(1, 5);
    auto subscriber = rx::make_subscriber<int>(fonnext, foncomplete);

    std::cout << "main thread Id: " << std::this_thread::get_id() << std::endl;
    std::cout << "just before subscribe " << std::endl;
    //  异步scheduler
    value$
        .map([](int v)
             {
                 std::cout << "map op thread Id: " << std::this_thread::get_id() << std::endl;
                return v; })
        .subscribe_on(threads)
        .observe_on(threads)
        .as_blocking() // 等待异步线程终止
        .subscribe(subscriber);
    std::cout << "just after subscribe " << std::endl;
}
